package com.lzy.gmall.realtime.app.dws;

import com.lzy.gmall.realtime.bean.T1;
import com.lzy.gmall.realtime.bean.T2;
import com.lzy.gmall.realtime.utils.MyClickHouseUtil;
import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *
 *
 *指标二 支付买家数
 *
 *
 */
public class Dws_Number_of_paid_buyers {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        TODO 2.设置并行度
        env.setParallelism(1);

        String topic = "dwd_order_pre";
        String groupId = "dws_payment_buyer_count";

        tableEnv.executeSql("CREATE TABLE DwsPaymentSource (" +
                "   user_id STRING," +
                "   order_id STRING," +
                "   order_status STRING," +
                "   create_time STRING," +
                "   operate_time STRING," +
                "   row_op_ts STRING," +
                "   source_type_name  STRING," +
//                 将 row_op_ts通过REPLACE转换为 Flink适用戳类型。
                "  `time_ltz` AS TO_TIMESTAMP_LTZ(" +
                "    UNIX_TIMESTAMP(REPLACE(row_op_ts, ' ', 'T'), 'yyyy-MM-dd''T''HH:mm:ss') * 1000," +
                "    3" +
                "  )," +
                "  WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND" +//设置水位线5秒延迟数据
                ")" + MyKafkaUtil.getKafkaDDL(topic, groupId));

//        通过子查询筛选出相关的订单状态
        Table t0 = tableEnv.sqlQuery("with  t1  as( " +
                "select   *   from DwsPaymentSource where order_status='1002' or order_status ='1004' or order_status='1005'  " +
                ") SELECT " +
                "  DATE_FORMAT(window_start, 'yyyy-MM-dd HH:mm:ss') AS stt," +
                "  DATE_FORMAT(window_end, 'yyyy-MM-dd HH:mm:ss') AS edt, " +
                "  COUNT(DISTINCT user_id) AS payment_buyers " +//去重支付买家数
                "FROM TABLE(" +
//                      滚动窗口每 10 秒生成一个不重叠的窗口统计窗口内的数据
                "  TUMBLE(TABLE t1, DESCRIPTOR(`time_ltz`), INTERVAL '10' SECOND)" +
                ") " +// 若过滤条件针对原始数据，使用WHERE
                "GROUP BY window_start, window_end");


        tableEnv.toAppendStream(t0, T2.class).addSink(MyClickHouseUtil.getSinkFunction("insert into gmall.Dws_Number_of_paid_buyers values(?,?,?)"));



        env.execute();
    }
}